package com.atguigu.dga.assessor.calc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.util.HttpUtil;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;

@Component("DATA_SKEW")
public class DataSkewAssessor extends Assessor {

    @Value("${spark.history.api.url}")
    private String spartURL;
    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {

        //排除ods层的数据
        if (assessParam.getTableMetaInfo().getTableMetaInfoExtra().getDwLevel().equals("ODS")){
            return;
        }
        //获取参数
        String metricParamsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject jsonObject = JSON.parseObject(metricParamsJson);
        Integer percent = jsonObject.getInteger("percent");
        Integer stageDurSeconds = jsonObject.getInteger("stage_dur_seconds");

        //获取作业id
        String yarnID = assessParam.getTDsTaskInstance().getAppLink();
        //排除无作业id的表
        if (yarnID==null || yarnID.trim().isEmpty()){
            governanceAssessDetail.setAssessComment("无任务id");
            return;
        }

        //获取作业尝试成功的id
        String attemptID = getAttempts(yarnID);

        //获取成功的阶段id
        List<String> stagesID = getStages(yarnID, attemptID);

        //获取各阶段的最大任务时间,平均任务时间,阶段时间
        List<parDate> testParDate = getParDate(yarnID, attemptID, stagesID);

        //遍历各阶段的时间,如果有超限的则0分
        for (parDate parDate : testParDate) {
            if (parDate.getMaxTaskDuration()>stageDurSeconds){
                Integer pertest=(parDate.getMaxTaskDuration()-parDate.getAvgTaskDuration())*100/ parDate.avgTaskDuration;
                if (pertest>percent){
                    governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
                    governanceAssessDetail.setAssessProblem("存在数据倾斜");
                }
            }
        }
        governanceAssessDetail.setAssessComment("testParDate: " + testParDate );
    }

    private List<parDate> getParDate(String yarnID, String attemptID, List<String> stagesID) {
        //保存最终数据的集合
        ArrayList<parDate> parDates = new ArrayList<>();

        //遍历所有阶段
        for (String stageId : stagesID) {
            String uri=spartURL+"/"+yarnID+"/"+attemptID+"/stages/"+stageId;
            String StagesString = HttpUtil.get(uri);
            //创建parDate
            parDate parDate = new parDate();
            //获取当前阶段json对象
            List<JSONObject> stageObjectList = JSONObject.parseArray(StagesString, JSONObject.class);
            //获取当前阶段任务信息
            for (JSONObject stageJsonObject : stageObjectList) {
                //保存所有任务的持续时间
                ArrayList<Integer> taskTime = new ArrayList<>();
                //保存最大持续时间
                Integer maxTime=0;
                //提取所有的任务
                JSONObject tasks = stageJsonObject.getJSONObject("tasks");
                //获取所有的key
                Set<String> keySet = tasks.keySet();
                //遍历key,获取所有的tesk对象
                for (String key : keySet) {
                    JSONObject tasksJSONObject = tasks.getJSONObject(key);
                    //判断成功的任务
                    if (tasksJSONObject.getString("status").equals("SUCCESS")){
                        //取该任务的持续时间
                        Integer duration = tasksJSONObject.getInteger("duration");
                        taskTime.add(duration);
                        maxTime=(maxTime>duration?maxTime:duration);
                    }
                }
                //求该阶段的平均时间
                Integer sumTime=0;
                for (Integer integer : taskTime) {
                    sumTime+=integer;
                }
                Integer avgTime=0;
                if (taskTime.size()==1){
                    avgTime=maxTime;
                }else {
                    avgTime=(sumTime-maxTime)/(taskTime.size()-1);
                }
                parDate.setAvgTaskDuration(avgTime);
                parDate.setMaxTaskDuration(maxTime);
            }
            parDate.setStageId(Integer.parseInt(stageId));
            parDates.add(parDate);
        }
        return parDates;
    }

    private List<String> getStages(String yarnID, String attemptID) {
        //保存成功阶段id集合
        ArrayList<String> strings = new ArrayList<>();
        String uri=spartURL+"/"+yarnID+"/"+attemptID+"/stages";
        String StagesString = HttpUtil.get(uri);
        List<JSONObject> jsonObjects = JSONObject.parseArray(StagesString, JSONObject.class);
        for (JSONObject jsonObject : jsonObjects) {
            if (jsonObject.getString("status").equals("COMPLETE")){
                strings.add(jsonObject.getString("stageId"));
            }
        }
        return strings;
    }

    private String getAttempts(String yarnID) {
        String uri=spartURL+"/"+yarnID;
        String attemptString = HttpUtil.get(uri);
        JSONObject parseObject = JSONObject.parseObject(attemptString);
        //获取阶段数组
        String attempts = parseObject.getString("attempts");
        List<JSONObject> jsonattempts = JSONObject.parseArray(attempts, JSONObject.class);
        for (JSONObject jsonattempt : jsonattempts) {
            if (jsonattempt.getString("completed").equals("true")){
                return jsonattempt.getString("attemptId");
            }
        }

        return null;
    }

    //返回值信息
    @Data
    private class parDate{
        //阶段id
        private Integer stageId ;
        //最大任务时间
        private Integer maxTaskDuration ;
        //平均任务时间
        private Integer avgTaskDuration ;

    }
}
